package com.thinking.kafkaTest;

import java.util.Arrays;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.kafka.BrokerHosts;
import org.apache.storm.kafka.KafkaSpout;
import org.apache.storm.kafka.SpoutConfig;
import org.apache.storm.kafka.StringScheme;
import org.apache.storm.kafka.ZkHosts;
import org.apache.storm.spout.SchemeAsMultiScheme;
import org.apache.storm.topology.TopologyBuilder;

public class StormTopology {

	
	public static void main(String[] args) {
		 // 注册 ZooKeeper 主机
        BrokerHosts brokerHosts = new ZkHosts(
                "192.168.7.133:2281");
        // 所接收 Kafka 的 topic 名称
        String topic = "yiweiOrder_mes";
        // ZooKeeper 的注册 node 名称（注意：需要加“/”，否则 ZooKeeper 会无法识别）
        String zkRoot = "/kafka_strom_order";
        
        // 配置 Spout
        String spoutId = "myKafka";
        SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, topic, zkRoot,
                spoutId);
      //启动之后不重新取kafka的数据
        spoutConfig.ignoreZkOffsets=false;
        
        spoutConfig.startOffsetTime = kafka.api.OffsetRequest.LatestTime();
        // 配置 Scheme（可选）
        spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
        spoutConfig.zkServers = Arrays.asList(new String[] {"192.168.7.133"});
        spoutConfig.zkPort = 2281;
        KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
        
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("kafka-stream", kafkaSpout);
        builder.setBolt("log4j-out-bolt", new JsonBolt()).shuffleGrouping("kafka-stream");
        Config config = new Config();
        config.setNumWorkers(1);
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("kafka-storm-yiweiorderMes", config,
                builder.createTopology());
	}
}
